package com.lagou.no1


import java.io.InputStream
import java.sql.{Connection, PreparedStatement, Statement}
import java.util.Properties

import com.alibaba.druid.pool.{DruidDataSource, DruidDataSourceFactory}
import javax.sql.DataSource
import org.apache.spark.sql.ForeachWriter

class MysqlWritte extends ForeachWriter[BusInfo]{
    //连接池
    var dataSource: DataSource = null
    //连接
    var conn:Connection = null
    //预写入对象
    var ps:PreparedStatement = null



    override def open(partitionId: Long, epochId: Long): Boolean = {
        val p:Properties = new Properties()
        val inputStream: InputStream = this.getClass.getResourceAsStream("/druid.properties")
        p.load(inputStream)
        //获取连接池
        dataSource = DruidDataSourceFactory.createDataSource(p)
        //获取连接
        conn = dataSource.getConnection()
        true
    }

    override def process(value: BusInfo): Unit = {
        //获取部署id和经纬度
        val deployNum: String = value.deployNum
        val lglat: String = value.lglat
        //插入mysql
        ps = conn.prepareStatement("insert into lglat_info values (?,?)")
        ps.setString(1,deployNum)
        ps.setString(2,lglat)
        ps.executeUpdate()
    }

    override def close(errorOrNull: Throwable): Unit = {
        //关闭对象
        ps.close()
        //关闭连接
        conn.close()

    }
}
